#include "rpcprovider.h"
#include "mprpcappliaction.h"
#include <functional>
#include "rpcheader.pb.h"
#include "logger.h"
#include "zookeeperutil.h"

void RpcProvider::NotifyService(::google::protobuf::Service *service)
{
    const google::protobuf::ServiceDescriptor *pserviceDesc = service->GetDescriptor();
    const std::string service_name = pserviceDesc->name();
    int methodCnt = pserviceDesc->method_count();

    ServiceInfo service_info;
    for (int i = 0; i < methodCnt; ++i)
    {
        const google::protobuf::MethodDescriptor *pmethodDesc = pserviceDesc->method(i);
        const std::string methodName = pmethodDesc->name();
        service_info.m_methodMap.insert({methodName, pmethodDesc});
    }
    service_info.m_service = service;
    m_serviceMap.insert({service_name, service_info});
}

void RpcProvider::Run()
{
    std::string ip = MprpcApplication::GetInstance().GetConfig().Load("rpcserverip");
    // MprpcApplication::GetConfig().Load("rpcserverip"); //因为是静态的，所以这样也可以
    uint16_t port = atoi(MprpcApplication::GetInstance().GetConfig().Load("rpcserverport").c_str());
    muduo::net::InetAddress address(ip, port);
    // std::unique_ptr<muduo::net::TcpServer> m_tcpserverPtr;
    muduo::net::TcpServer server(&m_evetloop, address, "rpcserver");
    server.setConnectionCallback(std::bind(&RpcProvider::OnConnection, this, std::placeholders::_1));
    server.setMessageCallback(std::bind(&RpcProvider::OnMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

    server.setThreadNum(4); // 1个连接线程，3个工作线程
    LOG_INFO("RpcProvider start service at ip: %s, port: %d", ip.c_str(), port);
    // std::cout << "RpcProvider start service at ip: " << ip << ", port: " << port << std::endl;

    // 把rppc服务发布到zk上，让rpc client可以从zk上发现服务，以服务为根节点，方法为子节点
    ZkClient zkCli;
    zkCli.Start();
    for (auto &sp : m_serviceMap)
    {
        std::string service_path = "/" + sp.first;
        zkCli.Create(service_path.c_str(), nullptr, 0);
        for (auto &mth_map : sp.second.m_methodMap)
        {
            std::string method_path = service_path + "/" + mth_map.first;
            char method_path_data[128] = {0};
            sprintf(method_path_data, "%s:%d", ip.c_str(), port);
            // 指明创建临时性节点
            zkCli.Create(method_path.c_str(), method_path_data, strlen(method_path_data), ZOO_EPHEMERAL);
        }
    }

    server.start();
    m_evetloop.loop();
}

void RpcProvider::OnConnection(const muduo::net::TcpConnectionPtr &conn)
{
    if (!conn->connected())
    {
        // rpc client断开连接
        conn->shutdown();
    }
}

/*
在框架内部需要RpcProvider和RpcConsumer之间的通信用的protobuf
service_name method_name args   定义message类型
UserServiceRpc Login zhangsan 123456

header_str + args_str
*/
void RpcProvider::OnMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buffer, muduo::Timestamp timestamp)
{
    // 网络上接受的远程rpc调用请求流
    std::string recv_buf = buffer->retrieveAllAsString();
    uint32_t header_size = 0;
    recv_buf.copy((char *)&header_size, 4, 0);

    std::string rpc_header_info = recv_buf.substr(4, header_size);
    mprpc::RpcHeader rpcHeader;
    std::string service_name;
    std::string method_name;
    uint32_t args_size;
    if (rpcHeader.ParseFromString(rpc_header_info))
    {
        service_name = rpcHeader.service_name();
        method_name = rpcHeader.method_name();
        args_size = rpcHeader.args_size();
    }
    else
    {
        LOG_ERROR("rpc header parase is error! input string is: %s", rpc_header_info.c_str());
        // std::cout << "rpc header parase is error! input string is: " << rpc_header_info << std::endl;
        return;
    }
    std::string args_str = recv_buf.substr(4 + header_size, args_size);

    LOG_INFO("======================================================");
    LOG_INFO("header_size: %d", header_size);
    LOG_INFO("rpc_header_info: %s", rpc_header_info.c_str());
    LOG_INFO("service_name: %s", service_name.c_str());
    LOG_INFO("method_name: %s", method_name.c_str());
    LOG_INFO("args_size: %d", args_size);
    LOG_INFO("args_str: %s", args_str.c_str());
    LOG_INFO("======================================================");
    // std::cout << "======================================================" << std::endl;
    // std::cout << "header_size: " << header_size << std::endl;
    // std::cout << "rpc_header_info: " << rpc_header_info << std::endl;
    // std::cout << "service_name: " << service_name << std::endl;
    // std::cout << "method_name: " << method_name << std::endl;
    // std::cout << "args_size: " << args_size << std::endl;
    // std::cout << "args_str: " << args_str << std::endl;
    // std::cout << "======================================================" << std::endl;

    auto sit = m_serviceMap.find(service_name);
    if (sit == m_serviceMap.end())
    {
        LOG_ERROR("%s is not exist!", service_name.c_str());
        // std::cout << service_name << " is not exist!" << std::endl;
        return;
    }
    auto mit = m_serviceMap[service_name].m_methodMap.find(method_name);
    if (mit == m_serviceMap[service_name].m_methodMap.end())
    {
        LOG_ERROR("%s is not exist!", method_name.c_str());
        // std::cout << method_name << " is not exist!" << std::endl;
        return;
    }
    google::protobuf::Service *service = sit->second.m_service;
    const google::protobuf::MethodDescriptor *method = mit->second;

    // 生成rpc调用的请求request和response
    google::protobuf::Message *request = service->GetRequestPrototype(method).New();
    if (!request->ParseFromString(args_str))
    {
        LOG_ERROR("request parse error, content: %s", args_str.c_str());
        // std::cout << "request parse error, content: " << args_str << std::endl;
        return;
    }
    google::protobuf::Message *response = service->GetResponsePrototype(method).New();

    google::protobuf::Closure *done = google::protobuf::NewCallback<RpcProvider, const muduo::net::TcpConnectionPtr &, google::protobuf::Message *>(this, &RpcProvider::SendRpcResponse, conn, response);
    service->CallMethod(method, nullptr, request, response, done);
}

void RpcProvider::SendRpcResponse(const muduo::net::TcpConnectionPtr &conn, google::protobuf::Message *response)
{
    std::string response_str;
    if (response->SerializeToString(&response_str))
    {
        // 序列化成功后，通过网络将rpc方法执行的结果返回给rpc调用方
        conn->send(response_str);
    }
    else
    {
        LOG_ERROR("response serialize error!");
        // std::cout << "response serialize error! content: " << std::endl;
    }
    // 模拟http的短连接服务，由rpcprovider主动断开连接，减少服务器保持连接的开销
    conn->shutdown();
}